package com.cike.sparkstudy.sparkstreaming.java;

import com.google.common.base.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * 使用UpdateStateByKey算子实现缓存实时WordCount
 */
public class UpdateStateByKeyWordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("UpdateStateByKeyWordCount").setMaster("local[2]");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        //使用UpdateStateByKey算子需要开启CheckPoint目录，开启checkpoint机制
        jssc.checkpoint("hdfs://bigdata:9000/wordcount_checkpoint");

        //在这里实现wordCount逻辑业务
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("bigdata", 9999);

        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });

        JavaPairDStream<String, Integer> pair = words.mapToPair(
                new PairFunction<String, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(String word) throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }
                });

        JavaPairDStream<String, Integer> wordCounts = pair.updateStateByKey(
                new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
                    @Override
                    //这里有两个参数，第一个是指这个batch中，这个key的新的值，可能有多个
                    //第二个参数是指之前的状态state,泛形自己制定
                    public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception {
                        //首先定义一个全局的计数单位
                        Integer newValue = 0;
                        //然后判断这个单词是否存在，不存在就是第一次出现
                        // 如果存在就说明之前已经统计过全局次数了
                        if (state.isPresent()) {
                            newValue = state.get();

                        }
                        // 接着把本次新出现的值都累加到newValue上,就是一个key目前的全局统计
                        for (Integer value : values) {
                            newValue += value;

                        }
                        return Optional.of(newValue);
                    }
                });
        //到这里都统计完了
        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();
        jssc.close();

    }
}
